package com.pw.study.flink.cep;

import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.List;
import java.util.Map;

/**
 * @Author: linux_future
 * @since: 2022/3/20
 **/
public class ExerciseCep_Combined extends BaseCEP{
    public static void main(String[] args) {
        new ExerciseCep_Combined().init();
    }

    @Override
    protected void handler(StreamExecutionEnvironment env, SingleOutputStreamOperator<JSONObject> stream) {
        SingleOutputStreamOperator<JSONObject> data = stream.map(new MapFunction<JSONObject, JSONObject>() {
            @Override
            public JSONObject map(JSONObject value) throws Exception {
                //System.out.println("==value: " + value);
                return value.getJSONObject("common");
            }
        });
       //data.print();
        // 定义模式
        Pattern<JSONObject, JSONObject> pattern = Pattern.<JSONObject>begin("start")
                .where(new SimpleCondition<JSONObject>() {
                    @Override
                    public boolean filter(JSONObject value) throws Exception {
                        return "Android 8.1".equals(value.getString("os"));
                    }
                }).where(new SimpleCondition<JSONObject>() {
                    @Override
                    public boolean filter(JSONObject value) throws Exception {
                        return "230000".equals(value.getString("ar"));
                    }
                }).or(new SimpleCondition<JSONObject>() {
                    @Override
                    public boolean filter(JSONObject value) throws Exception {
                        return "Android 11.0".equals(value.getString("os"));
                    }
                });
        Pattern<JSONObject, JSONObject> pt =mc();
        //绑定模式
        PatternStream<JSONObject> ps = CEP.pattern(data, pt);

        //获取结果
        ps.select(new PatternSelectFunction<JSONObject, String>() {
            @Override
            public String select(Map<String, List<JSONObject>> pattern) throws Exception {
                return pattern.toString();
            }
        }).print();

        ps.select(new PatternSelectFunction<JSONObject, String>() {
            @Override
            public String select(Map<String, List<JSONObject>> pattern) throws Exception {
                return pattern.get("start").toString();
            }
        }).print("other: ");

    }

    private Pattern<JSONObject, JSONObject> mc() {
        //模式匹配二
        Pattern<JSONObject, JSONObject> pattern = Pattern.<JSONObject>begin("start").where(new SimpleCondition<JSONObject>() {
            @Override
            public boolean filter(JSONObject value) throws Exception {
                return "Xiaomi 9".equals(value.getString("md"));
            }
        }).next("end").where(new SimpleCondition<JSONObject>() {
            @Override
            public boolean filter(JSONObject value) throws Exception {
                return "vivo iqoo3".equals(value.getString("md"));
            }
        });
        return pattern;
    }
}
